package cn.bytest.netty;

import cn.bytest.netty.code.NettyCodecAdapter;
import cn.bytest.netty.handler.NettyChannelHandler;
import cn.bytest.netty.handler.NettyClientHandler;
import cn.bytets.rpc.api.RpcClient;
import cn.bytets.rpc.common.URL;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.DefaultThreadFactory;
import org.apache.log4j.Logger;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import static java.util.concurrent.TimeUnit.MILLISECONDS;

public class NettyClient extends AbstractCodecAdapter implements RpcClient {

    private Logger logger = Logger.getLogger(NettyClient.class);

    private static int work = Math.min(Runtime.getRuntime().availableProcessors() + 1, 32);

    private static final NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup(work, new DefaultThreadFactory("NettyClientWorker", true));

    private Bootstrap bootstrap;

    private Channel channel;

    private URL url;

    private static Map<String,cn.bytets.rpc.api.Channel> channelMap = new ConcurrentHashMap<>();

    private volatile  boolean closed = false;

    private final Lock connectLock = new ReentrantLock();

    public NettyClient(URL url){
        this.url = url;
        openServer();
        connect(url.getHost(),url.getPort());
    }

    public void openServer(){
        final NettyClientHandler headler = new NettyClientHandler(new NettyChannelHandler());
        bootstrap = new Bootstrap();
        bootstrap.group(nioEventLoopGroup)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .channel(NioSocketChannel.class);
        /**
         * 链接超时
         */
        bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);
        bootstrap.handler(new ChannelInitializer() {

            @Override
            protected void initChannel(Channel ch) throws Exception {
                /**
                 * 心跳时间
                 */
                //int heartbeatInterval = UrlUtils.getHeartbeat(getUrl());
                NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec());
                ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
                        .addLast("decoder", adapter.getDecoder())
                        .addLast("encoder", adapter.getEncoder())
                        //.addLast("client-idle-handler", new IdleStateHandler(heartbeatInterval, 0, 0, MILLISECONDS))
                        .addLast("handler", headler);
//                String socksProxyHost = ConfigUtils.getProperty(SOCKS_PROXY_HOST);
//                if(socksProxyHost != null) {
//                    int socksProxyPort = Integer.parseInt(ConfigUtils.getProperty(SOCKS_PROXY_PORT, DEFAULT_SOCKS_PROXY_PORT));
//                    Socks5ProxyHandler socks5ProxyHandler = new Socks5ProxyHandler(new InetSocketAddress(socksProxyHost, socksProxyPort));
//                    ch.pipeline().addFirst(socks5ProxyHandler);
//                }
            }
        });
    }
    public void connect(String host,int port){
        connectLock.lock();
        try {
            ChannelFuture future = bootstrap.connect(host, port);
            boolean ret = future.awaitUninterruptibly(3000, MILLISECONDS);
            if (ret && future.isSuccess()) {
                System.out.println(String.format("%s:%s 链接成功", host, port));
                Channel newChannel = future.channel();
                try {
                    Channel oldChannel = NettyClient.this.channel;
                    if (null != oldChannel) {
                        try {
                            logger.info("关闭旧的链接:" + oldChannel + ",创建新的链接" + newChannel);
                            InetSocketAddress ipSocket = (InetSocketAddress) oldChannel.remoteAddress();
                            removeChannelToMap(getKey(ipSocket.getHostString(), ipSocket.getPort()));
                            oldChannel.close();
                        } finally {
                            NettyChannel.removeChannelIfDisconnected(oldChannel);
                        }
                    }
                    addChannelToMap(getKey(host, port), NettyChannel.getOrAddChannel(newChannel, null, host, port));
                } finally {
                    if (NettyClient.this.isClose()) {
                        newChannel.close();
                        NettyChannel.removeChannelIfDisconnected(newChannel);
                        removeChannelToMap(getKey(host, port));
                        NettyClient.this.channel = null;
                    } else {
                        NettyClient.this.channel = newChannel;
                    }
                }
            } else if (future.cause() != null) {
                throw new RuntimeException("链接服务端异常:" + host + ":" + port + "-->" + future.cause());
            } else {
                throw new RuntimeException("链接服务端异常:Client(" + host + ":" + port + ",链接超时");
            }
        } finally {
            connectLock.unlock();
        }
    }
    public void send(Object msg){
        if(!isConnect()){
            connect(url.getHost(),url.getPort());
        }
        if (null==channel||!channel.isActive()){
            throw new NullPointerException("发送消息:"+msg+" 但是 Client 为Null");
        }
        channel.writeAndFlush(msg);
    }
    public void close(){
        connectLock.lock();
        try {
            if (null != bootstrap) {
                nioEventLoopGroup.shutdownGracefully();
                bootstrap = null;
            }
            if (null != channel) {
                NettyChannel.removeChannelIfDisconnected(channel);
                InetSocketAddress address = getAddress();
                channelMap.remove(getKey(address.getHostName(), address.getPort()));
                channel.close();
                channel = null;
            }
            this.closed = true;
        }finally {
            connectLock.unlock();
        }
    }
    public boolean isClose(){
        return closed;
    }
    public boolean isConnect(){
        if (null==channel){
            return false;
        }
        return !isClose()&&channel.isActive();
    }
    private void addChannelToMap(String key,cn.bytets.rpc.api.Channel channel){
        channelMap.putIfAbsent(key,channel);
    }
    private void removeChannelToMap(String key){
        channelMap.remove(key);
    }
    private String getKey(String host,int port){
        return host+":"+port;
    }
    public cn.bytets.rpc.api.Channel getChannel(String host,int port){
        return getChannel(getKey(host,port));
    }
    private cn.bytets.rpc.api.Channel  getChannel(String key){
        return channelMap.get(key);
    }

    @Override
    public cn.bytets.rpc.api.Channel getChannel(URL url) {
        return getChannel(url.getHost(),url.getPort());
    }

    public InetSocketAddress getAddress(){
        if (null!=channel){
            return (InetSocketAddress)channel.remoteAddress();
        }
        return null;
    }
}
